package com.bungarus.busi;

import com.bungarus.Busi;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.*;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * The remedy lives at Busi node must send the non-ack message to the correct access node according to
 * the value of 'from' property in message.
 *
 * Created by tang on 2018/12/23.
 */
public class BusiRemedy extends Remedy {
    private Map<String, List<InnerClient>> accessClients;

    public static void start(BungarusServer server0) throws Exception{
        if(!started) {
            started = true;
            new BusiRemedy().run(server0);
        }
    }

    @Override
    protected void init(BungarusServer server){
        accessClients = ((Busi)server).getInnerClientList();
    }

    @Override
    protected void process() {
        DischargeQueue queue = Discharger.getQueueInstance();
        BusiRemedy.this.retry0(queue);
    }

    @Override
    protected void retry0(DischargeQueue queue) {
        logger.debug("Size of discharge queue: " + queue.size());
        if(queue.size() <= 0 || accessClients.isEmpty()){
            return;
        }
        ConcurrentHashMap.KeySetView<Long, MessageProtobuf.Message> keys = queue.keys();
        MessageProtobuf.Extension extension;
        for(Long l: keys){
            MessageProtobuf.Message noAckMsg = queue.getFromQueue(l);
            if((System.currentTimeMillis() - noAckMsg.getHeader().getSentTime()) >= ACK_TIMEOUT) {
                extension = noAckMsg.getHeader().getExtensions(0); // tag extension
                List<InnerClient> clients;
                String userId;
                if(extension.getValue().equalsIgnoreCase(Busi.CLUSTER_TAG_ACCESS)) {//sent to Access
                    clients = accessClients.get(Busi.CLUSTER_TAG_ACCESS);
                    if(MessageType.IM == noAckMsg.getHeader().getType()) {
                        userId = noAckMsg.getHeader().getTo();
                    }
                    else { //MessageType is IM_GROUP
                        userId = noAckMsg.getHeader().getExtensions(1).getValue();
                    }
                    Set sessions = SessionManagement.getInstance().retrieve(userId,
                            String.valueOf(noAckMsg.getHeader().getTenantID()));

                    sessions.stream().forEach(session -> {
                        String[] values = StringUtils.delimitedListToStringArray(session.toString(), SessionManagement.VALUE_SEPERATOR);
                        String reciptId = values[0];
                        String sessionId = values[1];
                        String ipport = values[2];
                        for(InnerClient client: clients) {
                            if(ipport.equals(client.getSrvIp() + ":" + client.getSrvPort())) {
                                try {
                                    if(MessageStatus.REMEDY == noAckMsg.getHeader().getReturnCode()) { // secondary delivering failed, persist it as RetryFailedMessage
                                        client.saveTryFailedMessage(noAckMsg);
                                    }
                                    else { // change messageType to Remedy and try again
                                        MessageProtobuf.Message tobeSent;
                                        if(MessageType.IM == noAckMsg.getHeader().getType()) {
                                            tobeSent = MessageBuilder.build2AccessMessage(Busi.CLUSTER_TAG_ACCESS, noAckMsg, sessionId);
                                        }
                                        else {
                                            tobeSent = MessageBuilder.buildGroupIMMessage(sessionId, reciptId, Busi.CLUSTER_TAG_ACCESS, noAckMsg);
                                        }
                                        client.sendMessage(MessageBuilder.buildRemedyMessage(tobeSent), Busi.CLUSTER_TAG_ACCESS);
                                    }
                                }
                                catch (BungarusClientException e) {
                                    e.printStackTrace();
                                    client.saveTryFailedMessage(noAckMsg);
                                }
                            }
                        }
                        queue.removeFromQueue(l);
                    });
                }
                else {//sent to busiBridge
                    clients = accessClients.get(Busi.CLUSTER_TAG_BUSI_BRIDGE);
                    long index = Long.parseLong(noAckMsg.getHeader().getId()) % clients.size();
                    InnerClient client = clients.get((int)index);
                    try {
                        if(MessageStatus.REMEDY == noAckMsg.getHeader().getReturnCode()) { // secondary delivering failed, persist it as RetryFailedMessage
                            client.saveTryFailedMessage(noAckMsg);
                        }
                        else {
                            client.sendMessage(MessageBuilder.buildRemedyMessage(noAckMsg), Busi.CLUSTER_TAG_BUSI_BRIDGE);
                        }
                        queue.removeFromQueue(l);
                    }
                    catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
}
